/*
 *  Queen Mary University of London MOM System
 *  
 *  Copyright Queen Mary University of London
 *  
 *  Authors:
 *  Bob Chew (bob.chew@elec.qmul.ac.uk)
 *  Beatriz Viñal Murciano (beavimu@gmail.com)
 *  John Bigham (john.bigham@elec.qmul.ac.uk)
 *  
 *  
 *  Portions of this software uses Apache Qpid
 *    http://qpid.apache.org
 *  
 *  Apache Qpid license:
 * --------------------------------------------------------------
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 * -------------------------------------------------------------
 * 
 */
package brokerFailureDetector;

import java.util.UUID;

import javax.jms.DeliveryMode;
import javax.jms.TopicSubscriber;

import org.apache.qpid.client.AMQConnection;
import org.apache.qpid.client.AMQSession;
import org.apache.qpid.client.AMQTopic;
import org.apache.qpid.client.TopicPublisherAdapter;
import org.apache.qpid.client.message.AbstractJMSMessage;
import org.apache.qpid.client.message.JMSTextMessage;

/**
 * Sends AMQP messages to itself through a broker and gets an idea of how well
 * that broker is working based on whether it gets the message or not.
 */
public class DetectorThread extends Thread {

    /** Detector. */
    private Detector bfDetector;

    /** Unique ID of the client. */
    private final String clientID;
    /** Name of the used topic. */
    private final String topicName;
    /** Machine in which the broker to test is running. */
    private final String testBroker;
    /** Port in which the broker to test is running. */
    private final int port;

    /** Time in which a message must arrive, in milliseconds. */
    private static final int MAX_WAIT = 10000;
    /** Time between messages. */
    private static final int INTERVAL = 10000;

    /** Best state of a broker. */
    private static final int BEST_STATE = 10;
    /** Worst state of a broker. */
    private static final int WORST_STATE = 0;

    /**
     * Constructor.
     *
     * @param bfDetector  to report the state of the broker to
     * @param testBroker  IP address of the broker to test
     * @param port  port in which the broker to test is listening
     */
    public DetectorThread(final Detector bfDetector, final String testBroker,
                          final int port) {

        this.bfDetector = bfDetector;
        clientID = UUID.randomUUID().toString();
        topicName = "brokerFailure." + clientID;
        this.testBroker = testBroker;
        this.port = port;

        start();

    }

    /**
     * Sends a message to the broker to test to a topic it is subscribed to and
     * waits for receiving the message.
     * If it arrives in MAX_WAIT milliseconds, the state of the broker is 10.
     * If it doesn't, the state of the broker is 0.
     * In any case, it tells the detector the state of the broker.
     * The process is repeated every INTERVAL milliseconds.
     */
    @Override
    public void run() {
        while (true) {
            try {

                //Create and start a connection
                String con = "amqp://" + bfDetector.getUSERNAME() + ":"
                             + bfDetector.getPASSWORD() + "@" + clientID + "/"
                             + bfDetector.getVIRTUAL_HOST()
                             + "?brokerlist='tcp://" + testBroker + ":"
                             + port + "'";
                AMQConnection connection = new AMQConnection(con);
                connection.start();
                //Create a session, non transacted and auto acknowledged
                AMQSession session = (AMQSession) connection.createSession(
                        false, AMQSession.AUTO_ACKNOWLEDGE);

                //Create topicURL
                String tURL = "topic://" + bfDetector.getEXCHANGE()
                              + "/?routingkey='" + topicName
                              + "'&exclusive='true'&autodelete='true'";
                //Create topic
                AMQTopic topic = (AMQTopic) session.createTopic(tURL);

                //Create a subscriber
                TopicSubscriber subscriber = session.createSubscriber(topic);
//                //Log
//              System.out.println("Thread in " + controller.getHOST_BROKER()
//                      + " testing " + TEST_BROKER + " subscribed to topic "
//                        + subscriber.getTopic());

                //Create publisher
                TopicPublisherAdapter publisher = (TopicPublisherAdapter)
                         session.createPublisher(topic);
                //Create the empty message to send
                JMSTextMessage txtMessage = (JMSTextMessage) session
                        .createTextMessage("");
                while (true) {
                    //Send the message created
                    publisher.publish(txtMessage, DeliveryMode.PERSISTENT,
                            AbstractJMSMessage.DEFAULT_PRIORITY,
                            AbstractJMSMessage.DEFAULT_TIME_TO_LIVE);
//                    //Log
//                    System.out.println("Thread in " + bfDetector.getHostIP()
//                            + " testing " + testBroker + " sent message to "
//                            + txtMessage.getJMSDestination());
                    //Receive a message if it arrives in the specified time
                    JMSTextMessage recMsg = (JMSTextMessage) subscriber
                            .receive(MAX_WAIT);
                    if (recMsg == null) {
                        throw new Exception(
                                "No message received in the specified time");
                    } else {
//                        System.out.println("Thread in "
//                                + bfDetector.getHostIP() + " testing "
//                                + testBroker + " received message \""
//                                + recMsg.getText() + "\" from "
//                                + recMsg.getJMSDestination());
                        //TODO: different numbers depending on the delay
//                        System.out.println("Thread in "
//                                + bfDetector.getHostIP() + " testing "
//                                + testBroker + ": OK");
                        bfDetector.updateState(testBroker, BEST_STATE);
                    }
                    //Sleep a bit
                    sleep(INTERVAL);
                }

            } catch (Exception e) {
//                e.printStackTrace();
//                System.err.println("Thread in " + bfDetector.getHostIP()
//                                   + " testing " + testBroker + ": FAIL");
                bfDetector.updateState(testBroker, WORST_STATE);
                try {
                    sleep(INTERVAL);
                } catch (InterruptedException e1) {
//                    e1.printStackTrace();
                }
            }
        }
    }

}
